package spring.cloud.tasks.common.zookeeper;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.Version;
import org.apache.zookeeper.*;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Slf4j
public class ZooKeeperManager {
    @Setter
    @Getter
    private String tasksId;
    @Setter
    @Getter
    private ZooKeeper zooKeeper;
    @Setter
    @Getter
    private List<ACL> aclList = new ArrayList<ACL>();
    @Setter
    @Getter
    private Properties properties;
    @Setter
    @Getter
    private boolean isCheckParentPathVersionIsCompatible = true;

    public ZooKeeperManager(Properties properties) throws Exception {
        this.properties = properties;
        this.tasksId = this.properties.getProperty(ZooKeeperManagerConstants.rootPath.toString());
        this.connect();//设置连接参数后就进行连接
    }

    /**
     * 检查当前以及父级路径不能被占用,lock file taskItemString:{tasksId}
     */
    public static void checkPathParentPathHasUsedByOtherProject(ZooKeeper zooKeeper, String path, String tasksId) throws Exception {
        String[] strings = path.split(ZookeeperNodeConstants.slash);
        String temporaryPath = "";
        for (int i = 0; i < strings.length - 1; i++) {
            String string = strings[i];
            if (string.equals("") == false) {
                temporaryPath = temporaryPath + ZookeeperNodeConstants.slash + string;
                //如果tasksId不为空或者一致则不允许连接操作。
                /**
                 * 专门建立一个lock文件标识该目录已经被tasks占用
                 */
                if (zooKeeper.exists(temporaryPath + ZookeeperNodeConstants.slash + "lock", false) != null) {
                    byte[] bytes = zooKeeper.getData(temporaryPath + ZookeeperNodeConstants.slash + "lock", false, null);
                    if (bytes != null) {
                        String value = new String(bytes);
                        if (!tasksId.equals(value)) {
                            throw new Exception("\"" + temporaryPath + "\"  is already a schedule instance's root directory, its any subdirectory cannot as the root directory of others");
                        }
                    }
                }
            }
        }
    }

    public String getRootPath() {
        return this.properties.getProperty(ZooKeeperManagerConstants.rootPath.toString());
    }

    public String getZooKeeperConnectString() {
        return this.properties.getProperty(ZooKeeperManagerConstants.zooKeeperConnectString.toString());
    }

    //
    private void connect() throws Exception {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        connectZookeeper(countDownLatch);
        countDownLatch.await(10, TimeUnit.SECONDS);
    }

    private void connectZookeeper(final CountDownLatch countDownLatch) throws Exception {
        zooKeeper = new ZooKeeper(//
                this.properties.getProperty(ZooKeeperManagerConstants.zooKeeperConnectString.toString()), //
                Integer.parseInt(this.properties.getProperty(ZooKeeperManagerConstants.zooKeeperSessionTimeout.toString())),//
                new Watcher() {//
                    public void process(WatchedEvent watchedEvent) {
                        sessionEvent(countDownLatch, watchedEvent);
                    }
                });
        String authString = this.properties.getProperty(ZooKeeperManagerConstants.userName.toString()) + ":" + this.properties.getProperty(ZooKeeperManagerConstants.password.toString());
        this.isCheckParentPathVersionIsCompatible = Boolean.parseBoolean(this.properties.getProperty(ZooKeeperManagerConstants.isCheckParentPathVersionIsCompatible.toString(), "true"));
        zooKeeper.addAuthInfo("digest", authString.getBytes());
        //权限初始化
        aclList.clear();
        aclList.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest(authString))));
        aclList.add(new ACL(ZooDefs.Perms.READ, Ids.ANYONE_ID_UNSAFE));
    }

    private void sessionEvent(CountDownLatch countDownLatch, WatchedEvent watchedEvent) {
        if (watchedEvent.getState() == KeeperState.NoSyncConnected) {
            try {
                reConnect();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        } else if (watchedEvent.getState() == KeeperState.SyncConnected) {
            countDownLatch.countDown();//ok
        } else if (watchedEvent.getState() == KeeperState.Disconnected) {
            try {
                reConnect();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        } else if (watchedEvent.getState() == KeeperState.Expired) {
            try {
                reConnect();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        } else {
            log.info("event.getState() =" + watchedEvent.getState() + ", event  value=" + watchedEvent.toString());
            countDownLatch.countDown();//状态待定
        }
    }

    public synchronized void reConnect() throws Exception {
        if (this.zooKeeper != null) {
            this.zooKeeper.close();
            this.zooKeeper = null;
            //
            this.connect();
        }
    }

    public void close() throws InterruptedException {
        log.info("close zookeeper");
        if (zooKeeper == null) {
            return;
        }
        this.zooKeeper.close();
    }

    public ZooKeeper getZooKeeper() throws Exception {
        if (this.zooKeeperIsConnected() == false) {
            reConnect();
        }
        return this.zooKeeper;
    }

    public boolean zooKeeperIsConnected() {
        return zooKeeper != null && zooKeeper.getState() == States.CONNECTED;
    }

    /**
     * zk连接成功后进行初始化
     */
    public void doInit() throws Exception {
        if (zooKeeper.exists(this.getRootPath(), false) == null) {
            ZooKeepers.createPath(zooKeeper, this.getRootPath(), aclList, CreateMode.PERSISTENT);
            checkPathParentPathHasUsedByOtherProject(zooKeeper, this.getRootPath(), tasksId);
            zooKeeper.setData(this.getRootPath(), spring.cloud.tasks.common.zookeeper.Version.getVersion().getBytes(), -1);
        } else {
            checkPathParentPathHasUsedByOtherProject(zooKeeper, this.getRootPath(), tasksId);
            byte[] bytes = zooKeeper.getData(this.getRootPath(), false, null);
            if (bytes == null) {
                zooKeeper.setData(this.getRootPath(), spring.cloud.tasks.common.zookeeper.Version.getVersion().getBytes(), -1);
            } else {
                String version = new String(bytes);
                if (spring.cloud.tasks.common.zookeeper.Version.isCompatible(version) == false) {
                    throw new Exception("tasks this version is:" + spring.cloud.tasks.common.zookeeper.Version.getVersion() + " ,tasks version in zk is:" + version);
                }
                log.info("" + Version.getVersion() + " " + version);
            }
        }
    }

}
